package com.demo.kafka.flink.test;
  
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
/*
* 打开终端，执行nc -l -p 9999，在 idea 运行以上代码，并在终端中依次发送下面的信息：

king 35.6
wang 36.3
zhao 36.7
lee 35.5
liu 37.0
kings 36.9
david 41.0
* */
public class FlinkSinkExample {
    public static void main(String[] args) throws Exception {  
        // 获取执行环境  
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
  
        // 从文件读取数据  
//        DataStream<String> data = env.readTextFile("E:\\github_project\\kafka\\src\\main\\java\\com\\demo\\kafka\\input\\words.txt");
        DataStream<String> data = env.socketTextStream("localhost", 9999);

        // 打印数据到控制台  
        data.print();  
  
        // 直接将文本数据写入另一个文本文件（注意：这通常不是Flink的推荐做法，因为它会尝试将所有数据写入单个文件）  
        // 这里仅为演示，实际应用中应避免这样做  
//        data.writeAsText("E:\\github_project\\kafka\\src\\main\\java\\com\\demo\\kafka\\output\\out.txt", FileSystem.WriteMode.OVERWRITE);
        data.print();

        // 将数据映射为元组并写入CSV文件  
        DataStream<Tuple3<String, String, String>> mappedData = data.map(new MapFunction<String, Tuple3<String, String, String>>() {  
            @Override  
            public Tuple3<String, String, String> map(String line) throws Exception {
                String[] arr = line.split(" ");  
                return new Tuple3<>(arr[0], arr[1], arr[2]);  
            }  
        });  
  
        // Flink的DataStream API本身不提供直接的writeAsCsv方法，因此这里使用自定义的Sink或写入到临时文本文件后转换  
        // 这里为了简单起见，我们使用writeAsText作为演示（但请注意，这不是CSV格式）  
        // 正确的CSV写入通常需要自定义Sink或使用第三方库  
//        mappedData.writeAsText("E:\\github_project\\kafka\\src\\main\\java\\com\\demo\\kafka\\output\\spilt.txt", FileSystem.WriteMode.OVERWRITE);
        mappedData.print();

        // 执行Flink作业  
        env.execute("Flink Sink Example");  
    }  
}  
  
// 注意：  
// 1. Flink的DataStream API默认不提供直接的writeAsCsv方法，因此上面的CSV写入只是演示，并不是真正的CSV格式。  
// 2. 为了写入CSV格式，你可能需要自定义Sink或使用如Apache Commons CSV等库来格式化数据。  
// 3. 上面的writeAsText到同一个目录（/home/shiyanlou/out/score.txt 和 /home/shiyanlou/out/score.csv）可能导致覆盖或命名冲突，这里仅为了演示。  
// 4. 确保你的Flink环境已经正确设置，包括依赖库等。